package dataflow.core;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.eclipse.emf.common.util.EList;

import dataflowScheme.ConnectionType;
import dataflowScheme.Port;
import dataflowScheme.PortSet;
import dataflowScheme.SynchroPort;

public class Task implements PE{

	static Logger log = Logger.getLogger(Task.class);
	
	/** source class on which processing will be called */
	private Class<?> source;
	/** source instance*/
	private Object sourceInstance;
	/** map of port set <-> method which will be called on source */
//	private Map<Integer, Method> runMethods = new HashMap<Integer, Method>();
	private Method runMethod;
	/** appropriate Task instance from input EMF scheme */
	private dataflowScheme.Task etask;
	
	/** map of port name <-> method of 'source' used to set data for port */
	private Map<String, Method> methods = new HashMap<String, Method>();
	/** map of port name <-> data which flow to this task */
	private Map<String, Object> values = new HashMap<String, Object>();
	/** map of port <-> number of port set */
	private Map<String, Integer> portsMap = new HashMap<String, Integer>();
	/** map of port set <-> current amount of data (when = 0 means need to run source) */
	private Map<Integer, Integer> portSetRate = new HashMap<Integer, Integer>();
	/** map of port set <-> amount of data necessary to execute task */
	private Map<Integer, Integer> portSetDefaultRate = new HashMap<Integer, Integer>();
	/** map of port name <-> connection anchor where data are coming 'from' or 'to' */
	private Map<String, ConnectionAnchor> oppossitePort = new HashMap<String, ConnectionAnchor>();
	/** port name <-> produced data token */
	private Map<String, Object> producedData = new HashMap<String, Object>();
	/** port name <-> signals to send */
	private List<String> signals2send = new ArrayList<String>();
	
	private Queue<Integer> iterationQueue = new LinkedList<Integer>();
	private final LinkedBlockingQueue<Event> eventQueue;
	private final LinkedBlockingQueue<ExecutionEvent> executionQueue;
	
	private static Class<?>[] setParams = new Class<?>[1];
	private static Class<?>[] getParams = new Class<?>[0];
	
	static{
		setParams[0] = java.lang.Object.class;
	}
	
	
	public Task(dataflowScheme.Task etask, LinkedBlockingQueue<Event> queue, LinkedBlockingQueue<ExecutionEvent> executionQueue){
		this.etask = etask;
		this.eventQueue = queue;
		this.executionQueue = executionQueue;
	}
	
	public boolean init() {//throws Exception{
		String className = getClassName(etask.getSource());
		Class<?>[] params = new Class<?>[1];
		params[0] = java.lang.Integer.class;
		boolean rv;
		
		try{
			source = Class.forName(className);
			runMethod = source.getMethod("run", params);

			rv = processMethods(etask.getPortSet());
			
		}catch(ClassNotFoundException cnfe){
			log.error("Task init interrupted - no class found", cnfe);
			return false;
		} 
		catch(NoSuchMethodException nsme){
			log.error("Task init interrupted - no 'run' method found", nsme);
			return false;
		}
		return rv;
		
	}
	
	private String getClassName(String resourceSource){
		String[] parts = resourceSource.split("/"); // /loadResource/model-src/dataflow/example/Worker2.java
		StringBuilder className = new StringBuilder();
		for(int i = 3; i < parts.length - 1; i++){
			className.append(parts[i]);
			className.append(".");
		}
		String fileName = parts[parts.length-1];
		className.append(fileName.split("[.]")[0]);
		
		log.debug(className.toString());
		
		return className.toString();
	}
	
	private boolean processMethods(EList<PortSet> portsets){
		
		boolean returnValue = true;
		
		for(int i = 0; i < portsets.size(); i++){
			int portsetNumber = i;
			PortSet ps = portsets.get(i);
			EList<Port> ports = ps.getPort();
			portSetRate.put(portsetNumber, ports.size());
			portSetDefaultRate.put(portsetNumber, ports.size());
			
			for(Port p : ports){
				String portName = p.getName();
				
				if(p instanceof SynchroPort){
					portsMap.put(portName, portsetNumber);
					continue;
				}
				
				if(methods.containsKey(portName)){
					log.error("Port name '" + portName + "' is duplicated.");
					returnValue = false;
					continue;
				}
				boolean in = p.getType().equals(ConnectionType.IN);
				String methodPrefix = in?"set":"get";
				String methodName = methodPrefix + portName;
				try{
					Method m = source.getMethod(methodName, in?setParams:getParams);
					methods.put(portName, m);
					portsMap.put(portName, portsetNumber);
				} catch (NoSuchMethodException nsme){
					log.error("Failed while looking for method '" + methodName + "' of '" + etask.getName() + "'", nsme);
					returnValue = false;
				}
			}
		}
		
		return returnValue;
	}
	
	private void run(int portset) {
		try{
			log.debug("starting executing '" + etask.getName() + "'");
			
			sourceInstance = source.newInstance();
			PortSet ps = etask.getPortSet().get(portset);
			if(ps == null){
				log.error("Bad number of portset.");
				return;
			}
		
			for (Port p : ps.getPort()){
				
				if (p instanceof SynchroPort){
					continue;
				}
				
				Object value = values.remove(p.getName());
				if(value == null){
					log.error("Sth wrong, can'r run portset when value doens't exist.");
					return;
				}
				try{
					methods.get(p.getName()).invoke(sourceInstance, value);
				}catch(InvocationTargetException ite){
					log.error("Invocation of port '" + p.getName() + "' failed.", ite);
					sourceInstance = null;
					return;
				}
				
			}
			
			executionQueue.put(new TaskExecutionEvent(runMethod, sourceInstance, portset, this));
//			Integer returnPortset = -1;
//			try{
//				returnPortset = (Integer)runMethod.invoke(sourceInstance, portset);
//			}catch(InvocationTargetException ite){
//				log.error("Invocation of main method failed.", ite);
//				sourceInstance = null;
//				return;
//			}
			
			
		} catch(IllegalAccessException ilae){
			log.error("Cannot create instance of Task", ilae);
		} catch(InstantiationException ie){
			log.error("Cannot create instance of Task", ie);
		} catch (InterruptedException e) {
			log.error("Execution queue is interuppted", e);
		}
	}

	private void sendResult(){
		for(Map.Entry<String, Object> e : producedData.entrySet()){
			sendData(e.getValue(), e.getKey());
		}

		Object[] signalPorts = signals2send.toArray();
		for(int i = 0; i < signalPorts.length; i++){
			String port = (String)signalPorts[i];
			sendSignal(port);
		}
	}
	
	private void sendData(Object data, String portName) {
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new DataEvent(out, data, in));
		} catch(InterruptedException ie){
			log.error("Trying to send data event when event queue is already dead.", ie);
		}
		
//		if(anchor.pe.receiveData(data, anchor.portName)){
//			//received successfully
//			producedData.remove(portName);
//			if(producedData.isEmpty() && signals2send.isEmpty()){
//				cleanForNextIteration();
//			}
//		}
	}
	
	private void sendSignal(String portName){
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new SynchroEvent(out, in));
		} catch(InterruptedException ie){
			log.error("Trying to send synchro event when event queue is already dead.", ie);
		}
		
//		if(anchor.pe.receiveSignal(anchor.portName)){
//			//received successfully
//			signals2send.remove(portName);
//			if(producedData.isEmpty() && signals2send.isEmpty()){
//				cleanForNextIteration();
//			}
//		}
	}

	private void cleanForNextIteration(){
		log.debug("cleaning for next iteration for '" + etask.getName() + "'");
	}
	
	@Override
	public boolean receiveData(Object data, String portName) {
		if(!portsMap.containsKey(portName)){
			log.error("Trying to set data to unknown port '" + portName + "'.");
			return false;
		}
		
		if(values.containsKey(portName)){
			//we have a token, therefore reject attempt of sender
			return false;
		}
		
		
		values.put(portName, data);
		int portsetNumber = portsMap.get(portName);
		int num = portSetRate.get(portsetNumber) - 1;
		
		if(num == 0){
			//time to make next iteration
//			portSetRate.put(portsetNumber, portSetDefaultRate.get(portsetNumber));
			invokeNextIteration(portsetNumber);
		}else{
			portSetRate.put(portsetNumber, num);
		}
		return true;
	}

	private void invokeNextIteration(int portSet){
		if(sourceInstance != null){
			//plan iteration
			iterationQueue.add(portSet);
		} else {
			run(portSet);
		}
	}
	
	@Override
	public void receiveResponse(EventMsg event, String portName) {
		if(event == EventMsg.freePort){
			if(producedData.containsKey(portName)){
				sendData(producedData.get(portName), portName);
			}
			return;
		}
		
		if(event == EventMsg.ok){
			if(producedData.containsKey(portName)){
				producedData.remove(portName);
			} else if(signals2send.contains(portName)){
				signals2send.remove(portName);
			}
			
			if(producedData.isEmpty() && signals2send.isEmpty()){
				cleanForNextIteration();
			}

		}
	}

	@Override
	public boolean receiveSignal(String portName) {
		if(!portsMap.containsKey(portName)){
			log.error("Trying to set data to unknown port '" + portName + "'.");
			return false;
		}
		
		if(values.containsKey(portName)){
			//we have a token, therefore reject attempt of sender
			return false;
		}
		
		int portsetNumber = portsMap.get(portName);
		int num = portSetRate.get(portsetNumber) - 1;
		
		if(num == 0){
			//time to make next iteration
//			portSetRate.put(portsetNumber, portSetDefaultRate.get(portsetNumber));
			invokeNextIteration(portsetNumber);
		}else{
			portSetRate.put(portsetNumber, num);
		}
		return true;
	}

	@Override
	public void addConnection(String portName,
			ConnectableEntity oppossiteEntity, String oppossitePortName) {
		oppossitePort.put(portName, new ConnectionAnchor(oppossiteEntity, oppossitePortName));
		
	}

	@Override
	public void receiveResult(int returnPortset) {
		try{
			if(returnPortset != -1){
				PortSet ps = etask.getPortSet().get(returnPortset);
				if(ps == null){
					log.error("Bad number of portset.");
					return;
				}
			
				for (Port p : ps.getPort()){
					
					if(p instanceof SynchroPort){
						signals2send.add(p.getName());
						continue;
					}
					Method getMethod = methods.get(p.getName());
					try{
						Object value = getMethod.invoke(sourceInstance, new Object[0]);
						producedData.put(p.getName(), value);
					}catch(InvocationTargetException ite){
						log.error("Invocation of port '" + p.getName() + "' failed.", ite);
						return;
					} catch (IllegalArgumentException e) {
						log.error("Method '" + getMethod.getName() + "' can't have any arguments.",e);
						return;
					} catch (IllegalAccessException e) {
						log.error("Method '" + getMethod.getName() + "' have to be public.",e);
						return;
					} 
					
				}
				
				sendResult();
			}
		} finally {
			sourceInstance = null;
		}
		
		Integer nextPortSet = iterationQueue.poll();
		if(nextPortSet != null){
			run(nextPortSet);
		}
		
	}
	
	public dataflowScheme.Task getETask(){
		return etask;
	}
	
}
